package com.bsj.power.pls.queue.service.impl.tactic;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.HexUtil;
import com.bsj.power.common.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @author yinhao
 * @version 1.0
 * @description 依次消费队列
 * @date 2024/10/18
 */
@Slf4j
public class SequentialConsumeQueue {

    //定长队列，不能超过一个字节的长度，否则无法正确组包
    private static ConcurrentLinkedQueue<byte[]> queue = new ConcurrentLinkedQueue<>();
    private static ReentrantReadWriteLock rwlock = new ReentrantReadWriteLock();

    //最新的星历数据
    private static byte[] ephByte = null;
//    private static boolean sendFlag = true;

    public boolean isRunning = true;
    private static Thread thread;
    //星历定时下发线程
    private static Thread ephThread;
    private static RTKConsumeHandle rtkConsumeHandle = null;

    public void doStart(Integer sleep) {
        rtkConsumeHandle = SpringUtil.getBean(RTKConsumeHandle.class);
        thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (isRunning) {
                    try {
                        byte[] bytes = queue.poll();
                        if (bytes != null) {
                            rtkConsumeHandle.handle(bytes);
                        }
                        Thread.sleep(sleep);
                    } catch (Exception e) {
                        log.error(ExceptionUtil.getMessage(e));
                        continue;
                    }
                }
            }
        });
        log.info("初始化顺序消费队列线程...");
        thread.start();
        ephThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (isRunning) {
                    try {
                        byte[] eph = getEphByte();
                        if (eph != null) {
                            putData(eph);
                        }
                        //每1分钟发送一次
                        Thread.sleep(60 * 1000);
                    } catch (Exception ex) {
                        log.error("星历定时下发线程异常: {}", ExceptionUtil.getMessage(ex));
                        continue;
                    }
                }
            }
        });
        log.info("初始化星历定时下发线程...");
        ephThread.start();
    }

    public void doStop() {
        isRunning = false;
        if (thread != null) {
            thread.interrupt();
        }
        if (ephThread != null) {
            ephThread.interrupt();
        }
    }

    /**
     * 依次写入数据
     *
     * @param bytes
     */
    public static void putData(byte[] bytes) {
        try {
            log.info(HexUtil.encodeHexStr(bytes));
            queue.add(bytes);
        } catch (Exception e) {
            log.error(ExceptionUtil.getMessage(e));
            return;
        }
    }

    /**
     * 依次消费数据
     *
     * @return
     */
//    public static byte[] takeData() {
//        try {
//            byte[] take = queue.take();
//            return take;
//        } catch (Exception e) {
//            log.error(ExceptionUtil.getMessage(e));
//            return null;
//        }
//    }

    /**
     * 需要检查队列是否已满
     *
     * @return
     */
    public static boolean checkSize() {
        int size = queue.size();
        if (size >= 254) {
            return false;
        } else {
            return true;
        }
    }

    public static byte[] getEphByte() {
        rwlock.readLock().lock();
        try {
            return ephByte;
        } finally {
            rwlock.readLock().unlock();
        }
    }

    public static void addEphByte(byte[] data) {
        rwlock.writeLock().lock();
        try {
            ephByte = data;
        } finally {
            rwlock.writeLock().unlock();
        }
    }
}
